#pragma once
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <iostream>
#include <memory>
#include <unordered_map>
#include <mutex>
#include <cstring>
#include <list>
namespace mymq
{
#define DATAFILE_SUBFIX ".mqd"
#define TMPFILE_SUBFIX ".mqd.tmp"
    using MessagePtr = std::shared_ptr<mymq::message>;
    // 消息数据的持久化管理
    class MessageMapper
    {
    public:
        MessageMapper(std::string &basedir, const std::string &qname) : _qname(qname)
        {
            // 判断父级目录最后一个字符是否为"/"
            if (basedir.back() != '/')
                basedir.push_back('/');
            _datafile = basedir + qname + DATAFILE_SUBFIX;
            _tmpfile = basedir + qname + TMPFILE_SUBFIX;
            // 创建父级目录
            if (FileHelper(basedir).exists() == false)
                assert(FileHelper::createderectory(basedir));
            CreateMsgFile();
        }
        bool CreateMsgFile()
        {
            // 存在不用创建
            if (FileHelper(_datafile).exists() == true)
                return true;
            // 不存在则创建
            bool ret = FileHelper::createfile(_datafile);
            if (ret == false)
            {
                DLOG("创建消息文件%s失败", _datafile.c_str());
                return false;
            }
            return true;
        }
        bool Insert(MessagePtr &msg)
        {
            return Insert(_datafile, msg);
        }
        void RemoveFile()
        {
            FileHelper::removefile(_datafile);
            FileHelper::removefile(_tmpfile);
        }
        // 删除某条消息
        bool Remove(MessagePtr &msg)
        {
            // 1.将该条消息的有效标志位标志为 "0"==删除 mutable_payload可修改 payload不可修改
            msg->mutable_payload()->set_valid("0");
            // 2.将数据序列化，重新加载（更新）到文件中
            //   1.如果重新更新数据时数据大小与原来数据大小不一致就不能再写入文件了
            std::string body = msg->payload().SerializeAsString();
            if (body.size() != msg->length())
            {
                DLOG("不能修改该数据，因为数据与原数据大小不一致");
                return false;
            }
            // 3.将数据写入到文件
            FileHelper helper(_datafile);
            bool ret = helper.write(body.c_str(), msg->offset(), body.size());
            if (ret == false)
            {
                DLOG("更新数据到文件失败");
                return false;
            }
            return true;
        }
        // 当文件消息超过2000条并且有效消息占有率<50%时需要清理无效数据
        std::list<MessagePtr> gc()
        {
            bool ret;
            // 1.先将有效消息加载到链表
            std::list<MessagePtr> result;
            ret = Locd(result);
            //DLOG("有效数据个数为 %ld", result.size());
            if (ret == false)
            {
                DLOG("加载有效消息失败");
                return std::list<MessagePtr>();
            }
            // 2.将数据备份到备份文件中
            //   1.创建备份文件
            //   2.将有效消息加载到文件中
            FileHelper::createfile(_tmpfile);
            for (auto &msg : result)
            {
                //DLOG("加载有效消息 %s", msg->payload().body().c_str());
                ret = Insert(_tmpfile, msg);
                if (ret == false)
                {
                    DLOG("加载有效消息到临时文件失败");
                    return result;
                }
            }
            //DLOG("临时文件大小 %ld", FileHelper(_tmpfile).size())
            // 3.将原文件删除
            ret = FileHelper::removefile(_datafile);
            if (ret == false)
            {
                DLOG("删除原文件失败");
                return result;
            }
            // 4.将临时文件的名字修改为原文件的名字
            ret = FileHelper(_tmpfile).rename(_datafile);
            if (ret == false)
            {
                DLOG("修改临时文件名字失败");
                return result;
            }
            //DLOG("新文件大小 %ld", FileHelper(_datafile).size())
            // 5.返回新的有效数据
            return result;
        }

    private:
        // 将有效数据加载到链表中
        bool Locd(std::list<MessagePtr> &result)
        {
            bool ret;
            FileHelper data_file_helper(_datafile);
            size_t offset = 0, msg_size;
            // 1.加载将该文件的所有有效数据
            while (offset < data_file_helper.size())
            {
                // （4字节）数据长度|数据内容
                ret = data_file_helper.read((char *)&msg_size, offset, sizeof(size_t));
                if (ret == false)
                {
                    DLOG("读取数据长度失败");
                    return false;
                }
                // 更新偏移量
                offset += sizeof(size_t);
                // 读取当前消息的数据
                std::string msg_body(msg_size, '\0');
                ret = data_file_helper.read(&msg_body[0], offset, msg_size);
                if (ret == false)
                {
                    DLOG("读取该条消息失败");
                    return false;
                }
                offset += msg_size;
                MessagePtr msgp = std::make_shared<message>();
                // 将该条消息反序列化到msgp中
                msgp->mutable_payload()->ParseFromString(msg_body);
                // 如果当前消息是无效的，直接跳过
                if (msgp->payload().valid() == "0")
                {
                    DLOG("加载到无效数据%s", msgp->payload().body().c_str());
                    continue;
                }
                // 将消息保存到链表
                result.push_back(msgp);
            }
            return true;
        }
        bool Insert(const std::string &filename, MessagePtr &msg)
        {
            // 1.将消息序列化，得到格式化的消息
            std::string body = msg->payload().SerializeAsString();
            // 2.获取文件长度==该条消息从这个位置开始写入
            FileHelper helper(filename);
            size_t fsize = helper.size();
            // 3.先写入4字节数据长度
            size_t len = body.size();
            bool ret = helper.write((char *)&len, fsize, sizeof(size_t));
            if (ret == false)
            {
                DLOG("写入数据长度失败");
                return false;
            }
            // 4.再将数据写入文件
            ret = helper.write(body.c_str(), fsize + sizeof(size_t), len);
            if (ret == false)
            {
                DLOG("写入数据到文件失败");
                return false;
            }
            // 5.更新msg中该消息的实际存储位置
            msg->set_offset(fsize + sizeof(size_t));
            msg->set_length(len);
            return true;
        }

    private:
        std::string _qname;    // 队列名字
        std::string _datafile; // 消息存储文件
        std::string _tmpfile;  // 备份消息文件
    };

    // 内存中消息的管理(以队列为单位)
    class QueueMessage
    {
    public:
        using Ptr = std::shared_ptr<QueueMessage>;
        QueueMessage(std::string &basedir, const std::string &qname)
            : _qname(qname),
              _mapper(basedir, qname),
              _valid_count(0),
              _total_count(0)
        {
        }
        // 恢复数据
        void Recovery()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _msgs = _mapper.gc(); // 获取当前队列所有的有效消息
            for (auto &msg : _msgs)
            {
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            // 更新数据数量
            _valid_count = _total_count = _msgs.size();
        }
        // 新增消息
        bool Insert(const BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            // 1.创建一条消息
            MessagePtr msg = std::make_shared<message>();
            msg->mutable_payload()->set_body(body);
            if (bp != nullptr) // 说明在上层已经设置好信息
            {
                // 如果该消息队列的数据是持久保存的那么这条消息的持久化信息由该条消息属性决定，队列不是持久化保存的那么该条消息也没必要持久化
                DeliveryMode mode = queue_is_durable ? bp->delivery() : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_delivery(mode);
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                // 如果未设置bp属性，那么使用默认的
                DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_delivery(mode);
                msg->mutable_payload()->mutable_properties()->set_id(GetUUIDHelper::GetUUID());
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            std::unique_lock<std::mutex> lock(_mutex);
            // 判断该条数据是否需要持久化
            if (msg->payload().properties().delivery() == DeliveryMode::DURABLE)
            {
                msg->mutable_payload()->set_valid("1"); // 将该消息设置为持久化
                bool ret = _mapper.Insert(msg);
                if (ret == false)
                {
                    DLOG("该消息持久化失败");
                    return false;
                }
                _valid_count += 1;
                _total_count += 1;
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            _msgs.push_back(msg);
            return true;
        }
        // 确认消息,在每次删除一个消息后判断是否需要清理文件
        bool Remove(const std::string &qid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _waitack_msgs.find(qid);
            if (it == _waitack_msgs.end())
            {
                // 没有找到该消息，不需要进行删除
                DLOG("没有找到该消息");
                return true;
            }
            // 找到了，根据是否为持久化消息进行对应的操作
            if (it->second->payload().properties().delivery() == DeliveryMode::DURABLE)
            {
                // 是持久化数据
                bool ret = _mapper.Remove(it->second);
                if (ret == false)
                {
                    DLOG("删除持久化消息失败%s", qid.c_str());
                    return false;
                }
                _durable_msgs.erase(qid);
                _valid_count -= 1;
                _total_count -= 1;
                // 进行判断是否需要清理文件，消息总量>2000并且有效消息占比小于50%
                gc();
            }
            _waitack_msgs.erase(qid);
            return true;
        }
        // 获取队首消息==推送一条消息到客户
        MessagePtr Front()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if(_msgs.size()==0)
            {
                DLOG("该队列已经没有消息");
                return MessagePtr();
            }
            if (_msgs.size() == 0)
            {
                return MessagePtr();
            }
            MessagePtr msg = _msgs.front();                                              // 推送一条数据到客户
            _msgs.pop_front();                                                           // 将推送的消息从待推送区移除
            _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(), msg)); // 添加到待确认消息中
            return msg;
        }
        // 待推送消息数
        size_t Getable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }
        // 获取该队列消息总量
        size_t Total_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }
        // 获取持久化的消息数量
        size_t Durable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }
        // 待回复消息数量
        size_t Waitack_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }
        void Clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.RemoveFile();
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            _valid_count = 0;
            _total_count = 0;
        }

    private:
        void gc()
        {
            // 如果消息还合法
            if (!(_total_count > 2000 && _valid_count * 10 / _total_count < 5))
                return;
            // 不合法
            // 1.从持久化文件中获取有效数据
            std::list<MessagePtr> msgs = _mapper.gc();
            // 2.从该更新的有效消息链表中遍历更新内存中的持久化消息链表
            for (auto &msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if (it == _durable_msgs.end())
                {
                    DLOG("垃圾回收后，有一条持久化消息没有在内存中被管理");
                    // 将该消息添加到内存
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                    _msgs.push_back(msg); // 添加到待处理链表末尾
                    continue;
                }
                // 更新每一条消息的位置
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }
            // 更新消息有关的数量
            _valid_count = _total_count = msgs.size();
        }

    private:
        std::string _qname;                                        // 队列名称
        MessageMapper _mapper;                                     // 持久化数据管理句柄，用来恢复数据
        size_t _valid_count;                                       // 有效消息总量
        size_t _total_count;                                       // 消息总量
        std::list<MessagePtr> _msgs;                               // 待推送消息链表
        std::unordered_map<std::string, MessagePtr> _durable_msgs; // 持久化消息
        std::unordered_map<std::string, MessagePtr> _waitack_msgs; // 待确认消息
        std::mutex _mutex;
    };
    // 内存中消息的管理
    class MessageManager
    {
    public:
        using Ptr = std::shared_ptr<MessageManager>;
        MessageManager(const std::string &basedir) : _basedir(basedir)
        {
        }
        // 初始化一条队列的消息
        void InitQueueMessage(const std::string &qname)
        {
            QueueMessage::Ptr qmp;
            // 为了防止双重加锁导致效率低下
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname); // 从map中查找是否存在该条队列记录
                if (it != _queue_msgs.end())       // 当前该队列已经被初始化了---存在了
                {
                    //DLOG("当前消息队列已经被初始化了");
                    return;
                }
                qmp = std::make_shared<QueueMessage>(_basedir, qname);
                _queue_msgs.insert(std::make_pair(qname, qmp));
            }
            qmp->Recovery(); // 恢复数据，原本就是带锁操作
        }
        // 删除该队列中的消息
        void DestroyQueueMessage(const std::string &qname)
        {
            QueueMessage::Ptr qmp;
            // 为了防止双重加锁导致效率低下
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname); // 从map中查找是否存在该条队列记录
                if (it == _queue_msgs.end())
                {
                    DLOG("不存在该消息队列");
                    return;
                }
                qmp = it->second;
                _queue_msgs.erase(qname);
            }
            qmp->Clear();
        }
        // 往队列中添加一条消息
        bool Insert(const std::string &qname, BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("没有找到该队列句柄%s", qname.c_str());
                    return false;
                }
                qmp = it->second;
            }
            return qmp->Insert(bp, body, queue_is_durable);
        }
        // 获取指定队列队头消息
        MessagePtr Front(const std::string &qname)
        {
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("没有找到该队列%s队首消息，未找到该消息管理队列句柄", qname.c_str());
                    return MessagePtr();
                }
                qmp = it->second;
            }
            return qmp->Front();
        }
        // 确认收到消息----删除消息
        void Ack(const std::string &qname, const std::string &msg_id)
        {
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("确认队列%s消息%s失败，未找到该消息管理队列句柄", qname.c_str(), msg_id.c_str());
                    return;
                }
                qmp = it->second;
            }
            qmp->Remove(msg_id);
        }
        size_t getable_count(const std::string &qname)
        {
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列%s待推送消息数量失败：没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->Getable_count();
        }
        size_t total_count(const std::string &qname)
        {
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列%s总持久化消息数量失败：没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->Total_count();
        }
        size_t durable_count(const std::string &qname)
        {
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列%s有效持久化消息数量失败：没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->Durable_count();
        }
        size_t waitack_count(const std::string &qname)
        {
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列%s待确认消息数量失败：没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->Waitack_count();
        }
        void Clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto &qmsg : _queue_msgs)
            {
                qmsg.second->Clear();
            }
        }

    private:
        std::mutex _mutex;
        std::string _basedir;
        std::unordered_map<std::string, QueueMessage::Ptr> _queue_msgs; // 某条队列对应的消息
    };
}